package tarns_result.old_version

import common.Tools.{ifDirExistsDelete, jsonFormatRECToCSYD_TABLEMETA_WORLD}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
//、、还没跑
object CSYD_TABLEMETA_WORLDJsonToREC {
  def main(args: Array[String]): Unit = {
    val conf=new SparkConf().setAppName("CSYD_TABLEMETA_WORLD").setMaster("local[*]")
    val spark=SparkSession.builder().config(conf).getOrCreate()
    val sc=spark.sparkContext
    //    val inputpath_sql = "/tmp/InternationalData_Result/CSYD_TABLEMETA_WORLD/"
    val outpath_result = "file:///Users/kelisiduofu/IdeaProjects/zhiwang_file/jiaoyu/CSYD_TABLEMETA_WORLDJsonToREC_RESULT"
    //val outpath_result="hdfs://node01:8020/tmp/InternationalData_Result/CSYD_TABLEMETA_WORLD_RESULT"
//    ifDirExistsDelete(outpath_result)
    //该路径下只有一个文件，但是该文件名不规则
    //6.将json串转化成REC格式(对应的中文列名)
    //    val inpaths: Array[String] = getHDFSDirFiles(inputpath_sql)
//    val dir="/Users/kelisiduofu/file/CSYD_TABLEMETA_WORLD/"
//    val file1=dir+"part-00000-a7977211-a1be-4346-a220-7d2d70b71b85-c000.json"
//    val file2=dir+"part-00001-a7977211-a1be-4346-a220-7d2d70b71b85-c000.json"
//    val file3=dir+"part-00002-a7977211-a1be-4346-a220-7d2d70b71b85-c000.json"
//    val file4=dir+"part-00003-a7977211-a1be-4346-a220-7d2d70b71b85-c000.json"
//    val file5=dir+"part-00004-a7977211-a1be-4346-a220-7d2d70b71b85-c000.json"
//    val file6=dir+"part-00005-a7977211-a1be-4346-a220-7d2d70b71b85-c000.json"
    val inpath="file:///Users/kelisiduofu/IdeaProjects/zhiwang_file/jiaoyu/CSYD_TABLEMETA_WORLD/part-00000-350c739a-45f1-4971-a2e6-fe2683e28cb5-c000.json"
    val json_file: RDD[String]=sc.textFile(inpath)

//    val inpaths: Array[String]=Array(file1,file2,file3,file4,file5,file6)
//    val json_file: RDD[String] = sc.textFile(inpaths.mkString(","))
    val REClines: RDD[String] = jsonFormatRECToCSYD_TABLEMETA_WORLD(json_file)
    //7.将结果写入到hdfs
//    REClines.repartition(5).saveAsTextFile(outpath_result)
    REClines.repartition(1).saveAsTextFile(outpath_result)
    sc.stop()
  }
}

